package com.hy.core.coreCode;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author: 王富贵
 * @description: 多线程合并消费逻辑
 * @createTime: 2022年08月10日 09:16:58
 */
@Slf4j
public class KillDemo {
    /**
     * 启动10个线程
     * 库存6个
     * 生成一个合并队列
     * 每个用户都能拿到自己的请求响应
     */
    //当前库存,实际上应该数据库拿
    private Integer stock = 6;
    //阻塞用户
    private BlockingQueue<RequestPromise> queue = new LinkedBlockingDeque<RequestPromise>(10);

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        KillDemo killDemo = new KillDemo();
        killDemo.mergeJob();
        Thread.sleep(2000);

        List<Future<Result>> futureList = new ArrayList<>();
        CountDownLatch countDownLatch = new CountDownLatch(10);
        for (int i = 0; i < 10; i++) {
            final Long orderId = i + 100L;
            final Long userId = Long.valueOf(i);
            Future<Result> future = executorService.submit(() -> {
                countDownLatch.countDown();

                countDownLatch.await(1000, TimeUnit.SECONDS);
                return killDemo.operate(new UserRequest(orderId, userId, 1));
            });
            futureList.add(future);
        }

        //结果响应这边，我们可以使用CompletedFuture进行谁先返回谁取得结果
        futureList.forEach(future -> {
            try {
                Result result = future.get(300, TimeUnit.MILLISECONDS);
                log.info("{}:客户端请求响应:{}", Thread.currentThread().getName(), result);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 用户库存扣减
     *
     * @param userRequest
     * @return
     */
    public Result operate(UserRequest userRequest) throws InterruptedException {
        //TODO 阈值判断
        //TODO 队列的创建

        //创建请求
        RequestPromise requestPromise = new RequestPromise(userRequest);

        /*
         * 简单解释一下为什么需要锁局部变量
         * 我们入队操作之后，再去调用wait方法的spike
         * 但是下面的轮询出队操作是异步线程取队列的，因此我们可能入队之后还未调用wait就执行出队并调用notify操作了
         * 在这种情况下，调用wait之后只能等待wait结束，也就是200毫秒才会得到结果，但这不是我们想要的
         *
         * 解决办法：
         * 采用局部变量锁起来，爆照wait调用在notify之前即可
         */
        synchronized (requestPromise) {
            boolean enqueueSuccess = queue.offer(requestPromise, 100, TimeUnit.MILLISECONDS);
            try {
                //如果入队失败
                if (!enqueueSuccess) {
                    return new Result(false, "系统繁忙");
                }
                //入队成功，用户等待200毫秒
                requestPromise.wait(200);
                //等待和中断都有可能唤醒，我们通过判断结果是否为空判断最终的执行结果
                if (requestPromise.getResult() == null) {
                    return new Result(false, "等待超时");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //正常情况正常返回
        return requestPromise.getResult();
    }

    //定时任务，循环
    public void mergeJob() {
        new Thread(() -> {
            //创建一个集合存放队列所有请求
            List<RequestPromise> promiseList = new ArrayList<>(10);
            //轮询消费
            while (true) {
                //当队列为空，等待10毫秒
                if (queue.isEmpty()) {
                    try {
                        Thread.sleep(10);
                        continue; // 一定要中断继续轮询
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                //队列不为空，将队列全部取出给到临时集合
                int queueSize = queue.size();
                for (int i = 0; i < queueSize; i++) {
                    promiseList.add(queue.poll());
                }

                //日志输出
                log.info("{}:合并扣减库存:{}", Thread.currentThread().getName(), promiseList);

                //获取需要扣减的库存,并判断是否够
                int sum = promiseList.stream().mapToInt(promise -> promise.getUserRequest().getCount()).sum();

                /*
                 * 库存够，直接扣减,继续轮询任务
                 * 库存不够，直接遍历，停止轮询任务
                 */
                if (sum <= stock) {
                    stock -= sum;
                    // TODO 数据库批量操作，建议先insert再update，节约update后insert还需要持有的锁
                    //通知用户
                    promiseList.forEach(requestPromise -> {
                        requestPromise.setResult(new Result(true, "ok"));
                        synchronized (requestPromise) {
                            requestPromise.notify();
                        }
                    });
                    promiseList.clear();
                    continue;
                }

                for (RequestPromise requestPromise : promiseList) {
                    int count = requestPromise.getUserRequest().getCount();
                    if (count <= stock) {
                        stock -= count;
                        // TODO 批量操作，建议先insert再update，节约update后insert还需要持有的锁
                        requestPromise.setResult(new Result(true, "ok"));
                    } else {
                        requestPromise.setResult(new Result(false, "库存不足"));
                    }
                    //针对库存不足进行通知
                    synchronized (requestPromise) {
                        requestPromise.notify();
                    }
                }
                promiseList.clear();
            }
        }, "mergeThread").start();
    }
}
